package com.sunday.common.mq.rabbit.study.base.e4_Routing;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Scanner;

/**
 * 路由
 * 在本教程中，我们将为其添加一个功能 - 我们将 使仅订阅消息的子集成为可能。为 例如，我们将只能将关键错误消息定向到 日志文件（以节省磁盘空间），同时仍然能够打印所有 控制台上的日志消息。
 * <p>
 * 直接交换
 * <p>
 * 我们上一个教程中的日志记录系统向所有消费者广播所有消息。我们希望扩展它以允许根据消息的严重性过滤消息。例如，我们可能需要一个程序将日志消息写入磁盘以仅接收严重错误，而不是在警告或信息日志消息上浪费磁盘空间。
 * 我们使用的是扇出交换，它没有给我们太大的灵活性 - 它只能进行无意识的广播。
 * 我们将使用直接交换。直接交换背后的路由算法很简单 - 消息进入队列，其  绑定密钥与消息的路由密钥完全匹配。
 * <p>
 * DIRECT("direct"),	路由模式
 * <p>
 * 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。
 * 这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “test”，
 * 则只有被标记为“test”的消息才被转发，不会转发test.aaa，也不会转发dog.123，
 * 只会转发test。
 * <p>
 * input -> error info warn
 */
@Slf4j
public class ReceiveLogsDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws Exception {

        ConnectionFactory factory = ConnectionUtil.factory();

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        Scanner scan = new Scanner(System.in);//实例化Scanner对象
        log.info("请输入文本:");
        String input = scan.nextLine();//接受用户输入的字符串

        argv = input.split(" ");

        if (argv.length < 1) {
            System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
            System.exit(1);
        }

        String queueName = channel.queueDeclare().getQueue();
        for (String severity : argv) {

            channel.queueBind(queueName, EXCHANGE_NAME, severity);

            log.info(" [*] Waiting for messages. To exit press CTRL+C");

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                log.info("[x] Received '{}':'{}'", delivery.getEnvelope().getRoutingKey(), message);
            };

            channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
            });

            if (severity.equals("error")) {
                push(channel, severity);
            }
        }

    }

    public static void push(Channel channel, String severity) throws IOException {

        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, severity);

        log.info(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            log.info("[x] Received '{}':'{}'", delivery.getEnvelope().getRoutingKey(), message);
        };

        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });
    }
}

